package com.changan.services;

import com.typesafe.config.Config;
import com.changan.enums.ConnectorType;


import java.util.Properties;

/**
 * @Author: wmh
 * Create Time: 2021/6/3 23:39
 */
public class FlinkPropertyService {
    public static Properties getFlinkProperties(Config config, ConnectorType type) {
        Properties properties = new Properties();
        config.entrySet().stream().filter(x ->
                x.getKey().startsWith(type.getValue() + ".") && (x.getValue() != null && (((String) x.getValue().unwrapped())).length() > 0)
        ).forEach(x -> properties.put(x.getKey().substring(type.getValue().length()+1), x.getValue().unwrapped()));
        return properties;
    }
}
